#include "Transceiver.h"

using namespace x2lib;

#pragma region Transceiver

Transceiver::Transceiver(bool isClient, uint8_t mgCode[16], int nPackSize):
    m_isClient(isClient)
{
    m_isRunning = false;
    m_pMyChannel = nullptr;
    memcpy(m_mgCode, mgCode, sizeof(m_mgCode));
}

Transceiver::~Transceiver()
{
}

bool Transceiver::Init()
{
    if (m_isClient)
    {
        m_pMyChannel = connectServer(false);
        if (m_pMyChannel != nullptr)
        {
            m_vecpChannel.push_back(m_pMyChannel);
        }
    }
    else
    {
        m_pMyChannel = prepareServer();
    }
    return (m_pMyChannel != nullptr);
}

bool Transceiver::Execute(int reConnWaitsec, int reConnTimes)
{
    if (m_isRunning || !m_pMyChannel) {
        return false;
    }
    m_isRunning = true;

    std::thread* pThread = nullptr;
    if (m_isClient)
    {
        if (m_pMyChannel)
        { 
            m_pMyChannel->conn = true;
            OnConnected(m_pMyChannel);
            m_mtxChannel.Lock();
            m_vecpChannel.push_back(m_pMyChannel);
            m_mtxChannel.Unlock();
        }
    }
    else
    {
        pThread = new std::thread([&]()-> void {
            do
            {
                AbsChannel* pChannel = acceptClients();
                if (pChannel)
                {
                    pChannel->conn = true;
                    OnConnected(pChannel);
                    m_mtxChannel.Lock();
                    m_vecpChannel.push_back(pChannel);
                    m_mtxChannel.Unlock();
                }
            } while (m_isRunning);
        });
    }

    uint8_t nIdle = 0;
    uint32_t index = 0;
    do
    {
        if (nIdle > 100 || m_vecpChannel.size() < 1)
        {
            std::this_thread::sleep_for(std::chrono::milliseconds(200));
            nIdle = 0;
        }
        ++nIdle;

        AbsChannel* pChannel = nullptr;
        m_mtxChannel.Lock();
        if (m_vecpChannel.size() > 0)
        {
            if (index >= m_vecpChannel.size())
            {
                index = 0;
            }
            pChannel = m_vecpChannel[index];
        }
        m_mtxChannel.Unlock();

        if (pChannel)
        {
            int nStatus1 = tryRecv(pChannel);
            int nStatus2 = trySend(pChannel, false);
            if (0 == nStatus1 || 0 == nStatus2)
            {
                pChannel->conn = false;
                OnConnected(pChannel);
                closeChannel(pChannel);
                m_mtxChannel.Lock();
                m_vecpChannel.erase(std::next(m_vecpChannel.begin(), index));
                m_mtxChannel.Unlock();
                --index;
            }
            else if (1 == nStatus1 && 1 == nStatus2)
            {
                nIdle = 0;
            }
            ++index;
        }
        else if (m_isClient)
        {
            while (--reConnTimes >= 0)
            {
                std::this_thread::sleep_for(std::chrono::seconds(reConnWaitsec));
                AbsChannel* pChannel = connectServer(true);
                if (pChannel)
                {
                    pChannel->conn = true;
                    OnConnected(pChannel);
                    m_mtxChannel.Lock();
                    m_vecpChannel.push_back(pChannel);
                    m_mtxChannel.Unlock();
                    break;
                }
            }
        }

    } while (m_isRunning);

    if (pThread)
    {
        pThread->join();
        delete pThread;
    }
    return 0;
}

void Transceiver::UnInit()
{

}

Transceiver::AbsChannel* Transceiver::FindChannel(std::function<bool(const AbsChannel*)> condition)
{
    AbsChannel* pChannel = nullptr;

    m_mtxChannel.Lock();
    for (auto it : m_vecpChannel)
    {
        if (condition(it)) 
        {
            pChannel = it;
            break;
        }
    }
    m_mtxChannel.Unlock();

    return pChannel;
}

bool Transceiver::PostData(AbsChannel* pChannel, const void* pData, uint32_t nData, uint32_t nMsgId, uint32_t iNodeI, const uint32_t iNodeX[4], bool isWait) 
{
    if (pChannel == nullptr) { return false; }
    bool isSucc = true;
    TranPack* pTranPack = (TranPack*)malloc(pChannel->uPackSize);
    for (uint32_t i = 0; i < nData; i += pChannel->uPackSize)
    {
        pTranPack->reset(m_mgCode);
        uint32_t nBytes = MIN(pChannel->uPackSize, nData - i);
        pTranPack->setPack(nMsgId, iNodeI, iNodeX, i + nBytes, nData, (uint8_t*)pData + i, (uint16_t)nBytes);

        if (!pushData((uint8_t*)pTranPack, TranPack::MIN_SIZE + pTranPack->Head.nBytes, pChannel, true, isWait)) 
        {
            isSucc = false;
            break;
        }
    }

    free(pTranPack);
    return isSucc;
}

bool Transceiver::SendData(AbsChannel* pChannel, const void* pData, uint32_t nData, uint32_t nMsgId, uint32_t iNodeI, const uint32_t iNodeX[4], bool isWait)
{
    if (!PostData(pChannel, pData, nData, nMsgId, iNodeI, iNodeX, isWait))
    {
        return false;
    }
    while(!pChannel->conn && pChannel->pPackTxBuffer->filledBytes() != 0);
    return pChannel->conn;
}

bool Transceiver::pushData(uint8_t* pData, uint32_t nData, AbsChannel* pChannel, bool isTx, bool isWait) 
{
    AbsChannel::Buffer* buffer = isTx ? pChannel->pPackTxBuffer : pChannel->pPackRxBuffer;
    
    while (true) 
    {
        if (nData <= buffer->spaceBytes()) 
        {
            break;
        }
        
        if (!isWait) { return false; }
        
        std::this_thread::sleep_for(std::chrono::milliseconds(100));
    }

    for (uint32_t i = 0; i < nData; ++i) 
    {
        (*buffer)[buffer->endIndex] = pData[i];
        buffer->endIndex = (uint16_t)(buffer->endIndex + 1) % buffer->totalBytes();
    }

    return true;
}

Transceiver::TranPack* Transceiver::pullData(AbsChannel* pChannel, bool isTx, bool isWait) 
{
    AbsChannel::Buffer* buffer = isTx ? pChannel->pPackTxBuffer : pChannel->pPackRxBuffer;
    TranPack* tranPack = isTx ? pChannel->pPackTxed : pChannel->pPackRxed;

    while (true) 
    {
        if (buffer->filledBytes() < TranPack::MIN_SIZE) 
        {
            if (!isWait) { break; }
            std::this_thread::sleep_for(std::chrono::milliseconds(100));
            continue;
        }

        // 查找魔术字位置
        int hashIdx = -1;
        for (uint32_t i = buffer->beginIndex; i != buffer->endIndex && hashIdx < 0; ++i) 
        {
            hashIdx = i;
			for (size_t j = 0; j < sizeof(m_mgCode); ++j)
			{
				if ((*buffer)[(i + j) % buffer->totalBytes()] != m_mgCode[j])
				{
					hashIdx = -1;
					break;
				}
			}
        }

        // 丢弃1字节
        if (hashIdx < 0) 
        {
            if (!isWait) { break; }
            buffer->beginIndex += 1;
            continue;
        }

        uint8_t* pData = (uint8_t*)tranPack;
        // 使头有效
        uint32_t bodyIdx = hashIdx;
        for (uint32_t j = 0; j < TranPack::MIN_SIZE; ++j) 
        {
            pData[j] = (*buffer)[bodyIdx];
            bodyIdx = (bodyIdx + 1) % buffer->totalBytes();
        }

        if (buffer->filledBytes() < TranPack::MIN_SIZE + tranPack->Head.nBytes)
        {
            if (!isWait) { break; }
            std::this_thread::sleep_for(std::chrono::milliseconds(100));
            continue;
        }
        buffer->beginIndex = bodyIdx;
        for (uint32_t j = 0; j < tranPack->Head.nBytes; ++j) 
        {
            pData[TranPack::MIN_SIZE+j] = (*buffer)[buffer->beginIndex];
            buffer->beginIndex = (buffer->beginIndex + 1) % buffer->totalBytes();
        }

        return tranPack;
    }

    return nullptr;
}

#if 0 // 示例（Socket）
int Transceiver::ApiSend(AbsChannel* pChannel, void* pData, int nData, int* pnBytes)
{
    int iRet = send(fd, (char*)pData, nData, 0);
    *pnByte = iRet > 0 ? iRet : 0;
#ifdef __X2LIB_WIN32__
    if (iRet < 1 && WSAEWOULDBLOCK != WSAGetLastError())
#elif defined __X2LIB_LINUX__
    if ((iRet == -1 && (errno == EBADF || errno == ECONNRESET)) // 主动关闭
        || (iRet == 0 && errno == EWOULDBLOCK) // 被动关闭
        || (iRet < 0 && (errno != EWOULDBLOCK) && (errno != EINPROGRESS))) // 正在接收
#endif
    {
        return false;
    }

    return true;
}

bool Transceiver::ApiRecv(void* pData, int nData, int* pnByte)
{
    // iRet=0 indicate socket closed,iRet<0 indicate send error
    int iRet = recv(fd, (char*)pData, nData, 0);
    *pnByte = iRet > 0 ? iRet : 0;
#ifdef __X2LIB_WIN32__
    if (iRet < 1 && WSAEMSGSIZE != WSAGetLastError())
#elif defined __X2LIB_LINUX__
    if ((iRet == -1 && (errno == EBADF || errno == ECONNRESET)) // 主动关闭
        || (iRet == 0 /*&& errno == EWOULDBLOCK*/) // 被动关闭
        || (iRet < 0 && (errno != EWOULDBLOCK) && (errno != EINPROGRESS))) // 正在接收
#endif
    {
        return false;
    }
    return true;
}
#endif

int Transceiver::trySend(AbsChannel* pChannel, bool isWait)
{
    int ret = 3;

    TranPack* pTranPack = pullData(pChannel, true, isWait);
    if (!pTranPack) { return ret; }

    uint8_t* pData = (uint8_t*)pTranPack;
    uint16_t nData = TranPack::MIN_SIZE + pTranPack->Head.nBytes;

    OnBytesEnSend(pData, nData);

    uint16_t index = 0;
    while (index < nData)
    {
        uint32_t nSent = 0;
        uint8_t result = ApiSend(pChannel, pData + index, nData - index, &nSent);
        if (result == 2 && isWait) 
        { 
            std::this_thread::sleep_for(std::chrono::milliseconds(50)); 
        }
        else if (result != 1)
        {
            ret = 0;
            break;
        }
        index += nSent;
    }

    if (nData == index) 
    { 
        ret = 1; 
    }
    else if (nData > index)
    {
        ret = 2;
    }

    OnBytesDeRecv(pData, nData);
    OnPackSended(pChannel);

    return ret;
}

int Transceiver::tryRecv(AbsChannel* pChannel)
{
    uint32_t nData = pChannel->pPackRxBuffer->spaceBytes();
    uint8_t* pData = (uint8_t*)malloc(nData);
    uint32_t nRecv = 0;
    bool result = ApiRecv(pChannel, pData, nData, &nRecv);
    if (!result) return 0;

    OnBytesDeRecv(pData, nRecv);

    pushData(pData, nRecv, pChannel, false, true);

    TranPack* pTranPack = pullData(pChannel, false, false);
    if (!pTranPack) return 2;

    OnPackRecved(pChannel);

    return 1;
}

#pragma endregion

